package com.gxl.butt.service.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutionException;

import com.alibaba.fastjson.JSONObject;
import com.gxl.butt.service.IkafkaOperatorService;
import com.gxl.butt.service.dto.Topic;
import com.gxl.butt.utils.JSONUtils;
import com.gxl.butt.KClientApplication;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.robert.kafka.kclient.boot.KClientBoot;

/**
 * Backend controller used to monitor and restart the KClient.
 *
 * @author Robert Lee
 * @since Aug 21, 2015
 *
 */

@RestController
public class KClientController {

	protected static Logger log = LoggerFactory.getLogger(KClientApplication.class);

	private ApplicationContext ctxKafkaProcessor = new ClassPathXmlApplicationContext("kafka-application.xml");

	@Autowired
	private IkafkaOperatorService kafkaOperatorService;

	@RequestMapping("/ButtNotice")
	public JSONObject ButtNotice(String token, String data, String topic) throws  Exception {

		JSONObject jd = new JSONObject();
		jd.put("data",data);
		JSONObject jsonObject = kafkaOperatorService.sendMes(topic, jd);
		Integer code = jsonObject.getInteger("error_code");
		if (code==0){
			return  JSONUtils.getRoot(0,"send success", "");
		}
		if (code==1){
			return  JSONUtils.getRoot(1,"该 topic 不存在", "");
		}
		return JSONUtils.getRoot(-1,"faild", "");
	}

	@RequestMapping("/createTopic")
	public JSONObject createTopic(@RequestBody Topic topic) {
		log.info("topic" + topic.getTopicname());
		JSONObject jsonObject = null;
		try {
			jsonObject = kafkaOperatorService.createTopic(topic);
		} catch (ExecutionException e) {
//			e.printStackTrace();
			return JSONUtils.getRoot(-2,e.getMessage(), "");
		} catch (InterruptedException e) {
//			e.printStackTrace();
			return JSONUtils.getRoot(-3,e.getMessage(), "");
		}
		Integer code = jsonObject.getInteger("error_code");
		if (code==0){
			return  JSONUtils.getRoot(0,"create success", "");
		}
		if (code==1){
			return  JSONUtils.getRoot(1,"该 topic 已经存在", "");
		}
		return JSONUtils.getRoot(-1,"faild", "");
	}


	@RequestMapping("/")
	public String hello() {
		return "Greetings from kclient processor!";
	}

	@RequestMapping("/status")
	public String status() {
		return "Handler Number: [" + getKClientBoot().getKafkaHandlers().size()
				+ "]";
	}

	@RequestMapping("/stop")
	public String stop() {
		log.info("Shutdowning KClient now...");
		getKClientBoot().shutdownAll();

		String startupTime = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss z")
				.format(new Date(ctxKafkaProcessor.getStartupDate()));
		log.info("KClient application stops at: " + startupTime);

		return "KClient application stops at: " + startupTime;
	}

	@RequestMapping("/restart")
	public String restart() {
		log.info("Shutdowning KClient now...");
		getKClientBoot().shutdownAll();

		log.info("Restarting KClient now...");
		ctxKafkaProcessor = new ClassPathXmlApplicationContext(
				"kafka-application.xml");

		String startupTime = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss z")
				.format(new Date(ctxKafkaProcessor.getStartupDate()));
		log.info("KClient application restarts at: " + startupTime);

		return "KClient application restarts at: " + startupTime;

	}

	private KClientBoot getKClientBoot() {
		return (KClientBoot) ctxKafkaProcessor.getBean("kClientBoot");
	}
}